#include "redis.hpp"
#include <iostream>

using std::cerr;
using std::endl;
using std::cout;

/*
redis作为集群服务器通信的基于发布-订阅消息队列时，会遇到两个难搞的bug问题，参考下面博客详细描述：
https://blog.csdn.net/QIANGWEIYUAN/article/details/97895611
*/

Redis::Redis():_publish_context(nullptr),_subscribe_context(nullptr){
    //构造函数初始化上下文指针为nullptr
}

Redis::~Redis(){
    //析构函数如果上下文指针不为空，则调用redis库函数释放
    //redisFree()为hiredis头文件里的函数

    if(_publish_context!=nullptr){
        redisFree(_publish_context);
    }

    if(_subscribe_context!=nullptr){
        redisFree(_subscribe_context);
    }
}

//连接redis服务器
bool Redis::connect(){
    //将两个上下文指针进行连接
    //连接方法也是调用redis库函数来执行
    _publish_context=redisConnect("127.0.0.1",6379);
    if(_publish_context==nullptr){
        cerr<<"Redis connect failed: publish_context redisConnect() error!"<<endl;
        return false;
    }

    _subscribe_context=redisConnect("127.0.0.1",6379);
    if(_subscribe_context==nullptr){
        cerr<<"Redis connect failed: subscribe_context redisConnect() error!"<<endl;
        return false;
    }

    //创建一个线程并分离，监听通道上的事件，有消息则给业务层上报
    //因为在Redis中调用subscribe是阻塞的，所以要专门开条线程执行等待通知
    std::thread subscribeThread([&](){
        observer_channel_message();
    });
    subscribeThread.detach();

    return true;
}

//向redis指定通道channel发送消息
bool Redis::publish(int channel,string message){
    //执行redis的publish发布消息命令，获取其结果集
    redisReply* reply=(redisReply*)redisCommand(_publish_context,"PUBLISH %d %s",channel,message.c_str());
    if(reply==nullptr){
        cerr<<"publish command failed!"<<endl;
        return false;
    }

    //释放结果集
    freeReplyObject(reply);

    return true;
}


//向redis指定的通道订阅消息
bool Redis::subscribe(int channel){
    // SUBSCRIBE命令本身会造成线程阻塞等待通道里面发生消息，这里只做订阅通道，不接收通道消息
    // 通道消息的接收专门在observer_channel_message函数中的独立线程中进行
    // 只负责发送命令，不阻塞接收redis server响应消息，否则和notifyMsg线程抢占响应资源

    /**
     * 从hiredis的redisCommand源码上可以看出，它实际上相当于调用了这三个函数：
     * redisAppendCommand 把命令写入本地发送缓冲区
     * redisBufferWrite 把本地缓冲区的命令通过网络发送出去
     * redisGetReply 阻塞等待redis server响应消息（真正阻塞的函数）
     * 
     * 所以我们这里只用调用前两个函数即可，第三个阻塞等待响应函数交给另外开辟线程运行
     */

    if(REDIS_ERR==redisAppendCommand(this->_subscribe_context,"SUBSCRIBE %d",channel)){
        cerr << "subscribe command failed!" << endl;
        return false;
    }

    // redisBufferWrite可以循环发送缓冲区，直到缓冲区数据发送完毕（done被置为1）
    //done在这里就是个标志位，记录缓冲区命令是否发送完毕
    int done=0;
    while(done==0){
        if(REDIS_ERR==redisBufferWrite(this->_subscribe_context,&done)){
            cerr << "subscribe command failed!" << endl;
            return false;
        }
    }

    return true;
}

// 向redis指定的通道unsubscribe取消订阅消息
bool Redis::unsubscribe(int channel){
    if (REDIS_ERR == redisAppendCommand(this->_subscribe_context, "UNSUBSCRIBE %d", channel)){
        cerr << "unsubscribe command failed!" << endl;
        return false;
    }

    // redisBufferWrite可以循环发送缓冲区，直到缓冲区数据发送完毕（done被置为1）
    int done = 0;
    while (done==0){
        if (REDIS_ERR == redisBufferWrite(this->_subscribe_context, &done)){
            cerr << "unsubscribe command failed!" << endl;
            return false;
        }
    }

    return true;
}


//在独立的线程中接收订阅的消息
void Redis::observer_channel_message(){
    /**
     * SUBSCRIBE某个通道时接收到消息例子如下：
     * 
     * 1) "message"
     * 2) "13"      //通道号
     * 3) "hello"   //消息正文
     */

    //redis命令结果集
    redisReply* reply=nullptr;
    
    //接收线程主循环，每次循环调用redisGetReply()阻塞接收消息
    while(REDIS_OK==redisGetReply(this->_subscribe_context,(void**)&reply)){
        if(reply->elements==0)
            continue;
        //结果集不为空，接收的消息正文不为空
        if(reply!=nullptr&&reply->element[2]!=nullptr&&reply->element[2]->str!=nullptr){
            //调用回调函数，将通道号和消息正文作为参数传输
            this->_notify_message_handler(atoi(reply->element[1]->str),reply->element[2]->str);
        }
        freeReplyObject(reply);
    }
    cerr << ">>>>>>>>>>>>> observer_channel_message quit <<<<<<<<<<<<<" << endl;
}


//初始化向业务层上报通道消息回调函数，设置回调函数
void Redis::init_notify_handler(std::function<void(int,string)> fn){
    //注册外界传过来的回调函数
    this->_notify_message_handler=fn;
}

